package cn.tedu.charging.cost.canal;

import cn.tedu.charging.common.pojo.po.ChargingCostRulePO;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import java.math.BigDecimal;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 计价规则数据 Canal 监听服务
 * 使用阿里巴巴官方 canal.client 实现数据同步
 */
@Slf4j
@Service
public class CostRuleCanalService {

    @Autowired
    private CanalConnector canalConnector;

    private ExecutorService executorService;

    private volatile boolean running = false;

    @PostConstruct
    public void start() {
        executorService = Executors.newSingleThreadExecutor();
        running = true;
        executorService.submit(new CanalWorker());
        log.info("计价规则 Canal 监听服务已启动");
    }

    @PreDestroy
    public void stop() {
        running = false;
        if (executorService != null) {
            executorService.shutdown();
        }
        log.info("计价规则 Canal 监听服务已停止");
    }

    private class CanalWorker implements Runnable {
        @Override
        public void run() {
            while (running) {
                try {
                    // 连接 Canal Server
                    canalConnector.connect();
                    // 订阅 charging_cost_rule 表
                    canalConnector.subscribe(".*\\.charging_cost_rule");
                    // 回滚到未 ack 的位置
                    canalConnector.rollback();

                    while (running) {
                        // 获取消息，每次最多获取1000条记录
                        Message message = canalConnector.getWithoutAck(1000);
                        long batchId = message.getId();
                        int size = message.getEntries().size();

                        if (batchId == -1 || size == 0) {
                            // 没有数据时休眠2秒
                            Thread.sleep(2000);
                        } else {
                            // 处理数据
                            processEntries(message.getEntries());
                            // 提交确认
                            canalConnector.ack(batchId);
                        }
                    }
                } catch (Exception e) {
                    log.error("Canal数据处理异常", e);
                    try {
                        Thread.sleep(5000); // 发生异常时休眠5秒再重试
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt();
                    }
                } finally {
                    canalConnector.disconnect();
                }
            }
        }
    }

    /**
     * 处理 Canal 条目数据
     * @param entries Canal 条目列表
     */
    private void processEntries(List<CanalEntry.Entry> entries) {
        for (CanalEntry.Entry entry : entries) {
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN 
                || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                continue;
            }

            try {
                CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                CanalEntry.EventType eventType = rowChange.getEventType();

                for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                    switch (eventType) {
                        case INSERT:
                            handleInsert(rowData);
                            break;
                        case UPDATE:
                            handleUpdate(rowData);
                            break;
                        case DELETE:
                            handleDelete(rowData);
                            break;
                        default:
                            break;
                    }
                }
            } catch (Exception e) {
                log.error("解析Canal数据异常", e);
            }
        }
    }

    @Autowired
    private RedisTemplate redisTemplate;

    /**
     * 处理插入操作
     * @param rowData 行数据
     */
    private void handleInsert(CanalEntry.RowData rowData) {
        ChargingCostRulePO costRule = convertToCostRule(rowData.getAfterColumnsList());
        log.debug("监听到计价规则新增数据:{}", costRule);
        log.debug("新增数据开始同步到Redis/ES/MQ");
        //通过RedisTemplate 写入数据
        //定义缓存的key  一个站点ID对应一个计价规则列表
        String key = "cost_rule_" + costRule.getStationId();
        //把旧数据删除 把缓存中旧的计价规则删除后,
        // 当计算价格接口被调用后,会 Redis中查询计价规则
        //发现计价规则被删除了,会重新查询数据库中最新的计价规则,从而实现了数据一致
        redisTemplate.delete(key);
        log.debug("数据库新增的计价规则数据Redis同步新增成功");
    }

    /**
     * 处理更新操作
     * @param rowData 行数据
     */
    private void handleUpdate(CanalEntry.RowData rowData) {
        ChargingCostRulePO before = convertToCostRule(rowData.getBeforeColumnsList());
        ChargingCostRulePO after = convertToCostRule(rowData.getAfterColumnsList());
        log.debug("监听到计价规则更新数据:before:{},after:{}", before, after);
        log.debug("更新数据开始同步到Redis");
        //通过RedisTemplate 更新数据

        //把旧数据删除 把缓存中旧的计价规则删除后,
        // 当计算价格接口被调用后,会 Redis中查询计价规则
        //发现计价规则被删除了,会重新查询数据库中最新的计价规则,从而实现了数据一致
        //redisTemplate.delete(key);

        // 数据库更新的时候 删除了Redis的旧数据 价格计算的时候 查询Redis 发现redis没有 查询数据库新数据
        // 更新到Redis 业务上 没问题 数据是一致的 缺点 会有一次数据的查询 只要修改价格 数据库的变化
        // 删了缓存 计算价格的时候 都需要 查询数据库最新的价格

        //思考 不删除 只更新 这样的好处是 减少数据库的查询
        //Redis String类型
        // key cost_rule_ + 站点ID
        // value 计价规则列表Json  [计价规则1(1.5),计价规则2(1.6),计价规则3(1.7)]
        // 管理员更新的时候 更新的是数据库的一条计价规则
        // redis 也对应的只更新一条
        //方案1>遍历
        // 查询Redis中对应的key 计价规则列表
        // 遍历列表
        // 找到具体的哪条变化了,然后把旧数据替换为新数据 再保存到Redis
        // 没问题 缺点 如果列表过长 遍历太慢 O(N) 时间复杂度 从Redis查询 保存数据到Redis 消耗网络的
        //方案?>
        // O(1) 时间复杂度 java中的数据结构 数组通过小标 Map O(1)
        // Redis K-V 缓存 底层是个大Map  O(1)  性能更高
        //用Redis的 Hash 存储计价规则 能不能?
        //key 和 value(Hash) 如何设计
        //key cost_rule_ + 站点ID  通过站点id 查询
        //value <Key value>
        // value key 计价规则的id
        // value value 计价规则PO

//        Map<String, Map<String,ChargingCostRulePO>> bigMap = new HashMap<>();
//
//        Map<String,ChargingCostRulePO> smallMa2 = bigMap.get(key);
//        ////更新小map中一条数据
//        smallMa2.put("计价规则的id1",new ChargingCostRulePO());
//        smallMa2.put("计价规则的id2",new ChargingCostRulePO());
//        bigMap.put("cost_rule_+站点ID-2", smallMa2);

        //定义缓存的key  一个站点ID对应一个计价规则列表
        String key = "cost_rule_hash_" + before.getStationId();
        //局部更新 通过 大key  cost_rule_+站点ID-2 和 小key  计价规则的id1 更新对应的 计价规则
        HashOperations<String,String,ChargingCostRulePO> hashOperations = redisTemplate.opsForHash();
        //put 不存在 创建 存在 更新
        //参数 大key 小key(hashKey field) value
        log.debug("更新Redis计价规则数据 key:{},hashKey:{},value:{}", key, after.getId(), after);
        hashOperations.put(key, String.valueOf(after.getId()), after);
        log.debug("更新Redis计价规则数据成功 key:{},hashKey:{},value:{}", key, after.getId(), after);
    }

    /**
     * 处理删除操作
     * @param rowData 行数据
     */
    private void handleDelete(CanalEntry.RowData rowData) {
        ChargingCostRulePO costRule = convertToCostRule(rowData.getBeforeColumnsList());
        log.debug("监听到计价规则删除数据:{}", costRule);
        log.debug("删除数据开始同步到Redis");
        //通过RedisTemplate 更新数据
        //定义缓存的key  一个站点ID对应一个计价规则列表
        String key = "cost_rule_" + costRule.getStationId();
        //把旧数据删除 把缓存中旧的计价规则删除后,
        // 当计算价格接口被调用后,会 Redis中查询计价规则
        //发现计价规则被删除了,会重新查询数据库中最新的计价规则,从而实现了数据一致
        redisTemplate.delete(key);
        log.debug("数据库删除的计价规则数据Redis同步删除成功");
    }

    /**
     * 将 Canal 列数据转换为 ChargingCostRulePO 对象
     * @param columns 列数据列表
     * @return ChargingCostRulePO 对象
     */
    private ChargingCostRulePO convertToCostRule(List<CanalEntry.Column> columns) {
        ChargingCostRulePO costRule = new ChargingCostRulePO();
        for (CanalEntry.Column column : columns) {
            switch (column.getName()) {
                case "id":
                    if (column.getValue() != null && !column.getValue().isEmpty()) {
                        costRule.setId(Integer.valueOf(column.getValue()));
                    }
                    break;
                case "station_id":
                    if (column.getValue() != null && !column.getValue().isEmpty()) {
                        costRule.setStationId(Integer.valueOf(column.getValue()));
                    }
                    break;
                case "gun_type":
                    if (column.getValue() != null && !column.getValue().isEmpty()) {
                        costRule.setGunType(Integer.valueOf(column.getValue()));
                    }
                    break;
                case "name":
                    costRule.setName(column.getValue());
                    break;
                case "start_time":
                    if (column.getValue() != null && !column.getValue().isEmpty()) {
                        costRule.setStartTime(Integer.valueOf(column.getValue()));
                    }
                    break;
                case "end_time":
                    if (column.getValue() != null && !column.getValue().isEmpty()) {
                        costRule.setEndTime(Integer.valueOf(column.getValue()));
                    }
                    break;
                case "power_fee":
                    if (column.getValue() != null && !column.getValue().isEmpty()) {
                        costRule.setPowerFee(new BigDecimal(column.getValue()));
                    }
                    break;
                case "service_fee":
                    if (column.getValue() != null && !column.getValue().isEmpty()) {
                        costRule.setServiceFee(new BigDecimal(column.getValue()));
                    }
                    break;
                default:
                    break;
            }
        }
        return costRule;
    }
}